package com.tr.rts.cdf

import java.io.ByteArrayOutputStream
import java.io._
import java.util.Date

import org.apache.avro.Schema
import org.apache.avro.generic.{GenericData, GenericDatumReader, GenericDatumWriter, GenericRecord}
import org.apache.avro.io._
import org.apache.commons.io.IOUtils
import org.scalatest.{FreeSpec, Matchers}

object avro {
  def Json2Schema(json: String): Schema = {
    new Schema.Parser().parse(json)
  }

  def xpath(record: GenericRecord, pathStr: String): Any = {
    val pathArray = pathStr.split("\\.")

    var r: GenericRecord = record
    for (step <- pathArray) {
      r = r.get(step).asInstanceOf[GenericRecord]
    }

    r
  }

  def Encode(record: GenericRecord, schema: Schema): Array[Byte] = {
    if(record == null || schema == null) return Array.emptyByteArray

    val out = new ByteArrayOutputStream
    val writer = new GenericDatumWriter[GenericRecord](schema)
    val encoder = EncoderFactory.get.binaryEncoder(out, null)
    writer.write(record, encoder)
    encoder.flush()
    out.flush()

    out.toByteArray
  }

  def Decode(schema: Schema, bytes: Array[Byte]): GenericData.Record = {
    val record = Decode(schema, schema, bytes)
    record
  }

  def Decode(schema: Schema, bytes: Array[Byte], offset: Int, length: Int): GenericData.Record = {
    val record = DecodeWithExtendSchema(schema, schema, bytes, offset, length)
    record
  }

  def Decode(recordSchema: Schema, extendSchema: Schema, bytes: Array[Byte]): GenericData.Record = {
    val record = DecodeWithExtendSchema(recordSchema, extendSchema, bytes, 0, bytes.length)
    record
  }

  def DecodeWithExtendSchema(recordSchema: Schema, extendSchema: Schema, bytes: Array[Byte], offset: Int, length: Int): GenericData.Record = {
    val reader = new GenericDatumReader[GenericRecord](recordSchema, extendSchema)
    val record = new GenericData.Record(extendSchema)
    if (length > 0) {
      val decoder = DecoderFactory.get.binaryDecoder(bytes, offset, length, null)
      reader.read(record, decoder)
    }
    record
  }
}

class cdfDataInspect extends FreeSpec with Matchers {

  val l2CommStr = getClass.getClassLoader.getResourceAsStream("l2Comm.avsc")
  val l2CommSchema = avro.Json2Schema(IOUtils.toString(l2CommStr))

  val fSnapshot = getClass.getClassLoader.getResourceAsStream("quote_snapshot.txt")
  val bSnapshot = ByteConverter.getRealBytes(IOUtils.toByteArray(fSnapshot))

  val registry = new BIMEcpRegistry()
  def parseRegistry(str: String): String = {
    str.contains("ecp:") match {
      case true =>
        val pos = str.indexOf("ecp:")
        val ecpid = str.substring(pos, pos + 42)
        val real = registry.toDescriptiveValue(ecpid)
        val newstr = str.replace(ecpid, real)
        parseRegistry(newstr)
      case _ => str
    }
  }

  val fdecode = (atime: Date) => {
    val l2j = CDFConsumer.decodeSnapshot(bSnapshot, atime)
    CDFConsumer.transform(l2j)
  }

  "atemporal" in {
    val content = fdecode(new Date())
    info(content)
    new PrintWriter("quote_atemporal.json") { write(content); close()}
  }

  val decoder = new sun.misc.BASE64Decoder()
  val fMifid = getClass.getClassLoader.getResourceAsStream("mifid1.txt")
  val mifid = decoder.decodeBuffer(fMifid)

  "property issue" in {
    val l2j = CDFConsumer.decodeSnapshot(mifid)
    val c = CDFConsumer.transform(l2j)

    info(c)
    new PrintWriter("44835155031.json") { write(c); close()}
  }

  /*"direct L2" in {
    val out = avro.Decode(l2CommSchema, bSnapshot)
    val content = out.toString
    info(content)

    val js = parse(content)
    val newjs = js.mapField { case (name, jv) =>
      val newval = jv match {
        case js: JString =>
          val str = js.values
          str.startsWith("ecp:") match {
            case true => JString(parseRegistry(str))
            case _ => js
          }
        case _ => jv
      }

      (name, newval)
    }

    val content1 = pretty(render(newjs))

    new PrintWriter("quote_fullhistory.json") { write(content1); close()}
  }

  "unitemporal" in {
    val now = new Date()
    val nowint = now.getTime().toString
    val c1 = fdecode(new Date(1518169997000L - 1))
    val c2 = fdecode(new Date(1518169997000L))
    val c3 = fdecode(new Date(1520917385742L))

    val outjson = s"""{ "systemtime": $nowint, "fullhistory": [{ "EffectiveFrom": 0, "data": $c1 }, { "EffectiveFrom": 1518169997000, "data": $c2 }, { "EffectiveFrom": 1520917385742, "data": $c3 }] } """
    val p = pretty(parse(outjson))
    info(p)

    new PrintWriter("quote_unitemporal.json") { write(p); close()}
  }*/
}
